package com.feriki.msync.process.sink;

import com.alibaba.fastjson.JSONObject;
import com.feriki.msync.constants.PolicyType;
import com.feriki.msync.constants.ConnectType;
import com.feriki.msync.meta.MilvusCollectionMeta;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.util.List;
import java.util.Optional;

/**
 * @author luozhida
 * @create 2024-07-10
 */
@Slf4j
public class JsonLineFileSinkHandler extends SinkHandler{
    public JsonLineFileSinkHandler(String filePath,String policy) {
        super(filePath, ConnectType.JSONL, policy);
    }

    @Override
    public boolean prepare(String source, MilvusCollectionMeta sourceMeta) {
        if(sourceMeta.getFieldTypes()==null || sourceMeta.getFieldTypes().isEmpty()){
            log.error("源{}字段列表为空", source);
            return false;
        }
        if(Files.exists(Paths.get(getName()),LinkOption.NOFOLLOW_LINKS)){
            if (PolicyType.COVER.equalsIgnoreCase(getPolicy())) {
                try {
                    Files.deleteIfExists(Paths.get(getName()));
                    log.info(String.format("%s模式触发,删除已存在的文件【%s】完成", PolicyType.COVER, getName()));
                } catch (IOException e) {
                    log.error(String.format("%s模式下,删除已存在的文件【%s】出现异常", PolicyType.COVER, getName()),e);
                    return false;
                }
            }else{
                try {
                    //读取文件第一行，约定是元数据
                    Optional<String> metaLine = Files.lines(Paths.get(getName()), StandardCharsets.UTF_8)
                            .filter(StringUtils::isNotBlank).findFirst();
                    if(metaLine.isPresent()) {
                        MilvusCollectionMeta sinkMilvusCollectionMeta = JSONObject.parseObject(metaLine.get(), MilvusCollectionMeta.class);
                        if (sourceMeta.getFieldTypes().size()!=sinkMilvusCollectionMeta.getFieldTypes().size()) {
                            log.error(String.format("源【%s】和文件【%s】的元数据不同，无法同步", source, getName()));
                            return false;
                        }
                    }else{
                        log.error("获取不到文件【{}】的元数据记录(约定首行数据是元数据)", getName());
                        return false;
                    }
                } catch (Exception ex) {
                    log.error("读取文件【{}】的元数据记录出现异常", getName(),ex);
                    return false;
                }
            }
        }

        //不存在文件则创建文件
        if(!Files.exists(Paths.get(getName()),LinkOption.NOFOLLOW_LINKS)){
            try {
                if(Paths.get(getName()).getParent()!=null && !Files.exists(Paths.get(getName()).getParent())){
                    Files.createDirectories(Paths.get(getName()).getParent());
                }
                Files.createFile(Paths.get(getName()));
            } catch (IOException e) {
                log.error(String.format("创建文件【%s】出现异常", getName()),e);
                return false;
            }
        }
        try {
            Path filePath = Paths.get(getName());
            if(filePath.toFile().length() == 0 ){
                //当文件为空时，首行写入元数据
                String metaContent = (JSONObject.toJSONString(sourceMeta)+"\n");
                Files.write(filePath, metaContent.getBytes(StandardCharsets.UTF_8), StandardOpenOption.APPEND);
            }
        } catch (IOException e) {
            log.error(String.format("向文件【%s】写入元数据出现异常", getName()),e);
            return false;
        }
        return true;
    }

    @Override
    public long handler(String source, List<JSONObject> sourceJsonObjects) {
        if(sourceJsonObjects==null || sourceJsonObjects.isEmpty()) {
            return 0;
        }
        StringBuffer writtenValueSb = new StringBuffer();
        sourceJsonObjects.forEach(sourceJsonObject->
                writtenValueSb.append(sourceJsonObject.toJSONString()).append("\n")
        );
        try {
            Files.write(Paths.get(getName()), writtenValueSb.toString().getBytes(StandardCharsets.UTF_8), StandardOpenOption.APPEND);
            log.info(String.format("当前成功从源【%s】同步到文件【%s】%s条记录", source, getName(), sourceJsonObjects.size()));
            return sourceJsonObjects.size();
        } catch (IOException e) {
            log.error(String.format("当前从源【%s】同步到文件【%s】%s条记录失败", source, getName(), sourceJsonObjects.size()),e);
        }
        return 0;
    }

    @Override
    public void destroy() {

    }
}
